package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSQLApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeOrderDetail().start(
                10014,
                4,
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        //从kafka的topic主题中读取数据，创建动态表
        readOdsDB(tableEnv,Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);

        //过滤出订单明细表
        Table detailTable = tableEnv.sqlQuery("select \n" +
                "data['id'] `id`,\n" +
                "data['course_id'] `course_id`,\n" +
                "data['course_name'] `course_name`,\n" +
                "data['order_id'] `order_id`,\n" +
                "data['user_id'] `user_id`,\n" +
                "data['origin_amount'] `origin_amount`,\n" +
                "data['coupon_reduce'] `coupon_reduce`,\n" +
                "data['final_amount'] `final_amount`,\n" +
                "data['create_time'] `create_time`,\n" +
                "ts\n" +
                "from topic_db\n" +
                "where `table` = 'order_detail'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("order_detail",detailTable);
        //tableEnv.executeSql("select * from order_detail").print();

        //过滤出订单数据
        Table orderInfo = tableEnv.sqlQuery("select\n" +
                "data['id'] `order_info_id`,\n" +
                "data['order_status'] `order_status`,\n" +
                "data['out_trade_no'] `out_trade_no`,\n" +
                "data['trade_body'] `trade_body`,\n" +
                "data['session_id'] `session_id`,\n" +
                "data['province_id'] `province_id`\n" +
                "from topic_db\n" +
                "where `table` = 'order_info'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("order_info",orderInfo);
        //tableEnv.executeSql("select * from order_info").print();

        //对两张表进行关联
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "od.`id`,\n" +
                "od.`course_id`,\n" +
                "od.`course_name`,\n" +
                "od.`order_id`,\n" +
                "od.`user_id`,\n" +
                "od.`origin_amount`,\n" +
                "od.`coupon_reduce`,\n" +
                "od.`final_amount`,\n" +
                "od.`create_time`,\n" +
                "date_format(od.`create_time`,'yyyy-MM-dd') `date_id`,\n" +
                "oi.`order_status`,\n" +
                "oi.`out_trade_no`,\n" +
                "oi.`trade_body`,\n" +
                "oi.`session_id`,\n" +
                "oi.`province_id`,\n" +
                "od.ts\n" +
                "from order_detail od\n" +
                "join order_info oi on od.order_id = oi.order_info_id");
        tableEnv.createTemporaryView("order_detail_info",resultTable);
        //tableEnv.executeSql("select * from order_detail_info").print();

        //将连接的结果写入主题进行映射
        //创建动态表和要写入的主题进行映射
        tableEnv.executeSql("create table dwd_trade_order_detail(\n" +
                "id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "create_time string,\n" +
                "date_id string,\n" +
                "order_status string,\n" +
                "out_trade_no string,\n" +
                "trade_body string,\n" +
                "session_id string,\n" +
                "province_id string,\n" +
                "ts bigint,\n" +
                "primary key(id) not enforced\n" +
                ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));

        //写入
        resultTable.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL);

    }
}
